package wh.coach.utils;

import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.TimeoutUtils;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.util.Assert;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @author hangwen
 * @date 2023/12/13
 */
public class RedisObjectStore<V> implements ObjectStore<V> {

    private static final String NOT_EXISTS = "$NOT_EXISTS";

    private final StringRedisTemplate redisTemplate;
    private final Class<V> type;

    public RedisObjectStore(final StringRedisTemplate redisTemplate, Class<V> type) {
        this.redisTemplate = redisTemplate;
        this.type = type;
    }

    @Override
    public ObjectWrapper<V> getObject(String key) {
        Assert.hasText(key, "parameter 'key' is required");

        final String json = this.redisTemplate.opsForValue().get(key);
        if (json == null) {
            return ObjectWrapper.unknownObject();
        }

        if (NOT_EXISTS.equals(json)) {
            return ObjectWrapper.notExistsObject();
        }

        return ObjectWrapper.of(JsonUtils.toJavaObject(json, this.type));
    }

    @Override
    public List<ObjectWrapper<V>> getObjects(final Collection<String> keys) {
        Assert.notEmpty(keys, "parameter 'keys' is required");


        List<ObjectWrapper<V>> objectWrappers = new ArrayList<>(keys.size());

        //转换成流式然后进行一个返回前的比较
        final Stream<String> stream = keys.stream();
        List<Object> jsonList = stream
                .map(key -> {
                    final String json = this.redisTemplate.opsForValue().get(key);
                    if (json == null) {
                        //那就是不存在，返回not_exists
                        return NOT_EXISTS;
                    }
                    //存在返回json
                    return json;
                })
                .collect(Collectors.toList());

        //遍历集合jsonlist转换为ObjectWrapper以告知接口返回的状态
        for (Object json : jsonList) {
            if (NOT_EXISTS.equals(json)) {
                objectWrappers.add(ObjectWrapper.notExistsObject());
            } else {
                V value = JsonUtils.toJavaObject((String) json, this.type);
                objectWrappers.add(ObjectWrapper.of(value));
            }
        }

        return objectWrappers;
    }

    @Override
    public void remove(String key) {
        this.redisTemplate.delete(key);
    }

    @Override
    public void setNotExists(final String key) {
        Assert.hasText(key, "parameter 'key' is required");

        this.redisTemplate.opsForValue().set(key, NOT_EXISTS);
    }

    @Override
    public void setNotExists(final String key, final long timeout, TimeUnit unit) {
        Assert.hasText(key, "parameter 'key' is required");
        Assert.state(timeout > 0, "parameter 'timeout' must be > 0");
        Assert.notNull(unit, "parameter 'unit' is required");

        this.redisTemplate.opsForValue().set(key, NOT_EXISTS, timeout, unit);
    }

    @Override
    public void setObject(String key, V object) {
        Assert.hasText(key, "parameter 'key' is required");
        Assert.notNull(object, "parameter 'value' is required");

        String json = JsonUtils.toJsonString(object);
        this.redisTemplate.opsForValue().set(key, json);
    }

    @Override
    public void setObject(String key, V object, long timeout, TimeUnit unit) {
        Assert.hasText(key, "parameter 'key' is required");
        Assert.notNull(object, "parameter 'value' is required");
        Assert.state(timeout > 0, "parameter 'timeout' must be > 0");
        Assert.notNull(unit, "parameter 'unit' is required");

        String json = JsonUtils.toJsonString(object);
        this.redisTemplate.opsForValue().set(key, json, timeout, unit);
    }

    @Override
    public void setObjects(Collection<String> keys, Collection<V> objects) {
        Assert.notEmpty(keys, "parameter 'keys' is required");
        Assert.notEmpty(objects, "parameter 'objects' is required");
        Assert.state(keys.size() == objects.size(), "size of 'keys' and 'objects' must be the same");

        List<String> jsonList = objects.stream()
                .map(JsonUtils::toJsonString)
                .toList();

        this.redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            final RedisSerializer<String> stringSerializer = this.redisTemplate.getStringSerializer();

            Iterator<String> keysIterator = keys.iterator();
            Iterator<String> jsonIterator = jsonList.iterator();

            while (keysIterator.hasNext() && jsonIterator.hasNext()) {
                final String key = keysIterator.next();
                final String json = jsonIterator.next();

                final byte[] rawKey = stringSerializer.serialize(key);
                final byte[] rawValue = stringSerializer.serialize(json);

                connection.set(rawKey, rawValue);
            }

            return null;
        });
    }

    @Override
    public void setObjects(Collection<String> keys, Collection<V> objects, long timeout, TimeUnit unit) {
        Assert.notEmpty(keys, "parameter 'keys' is required");
        Assert.notEmpty(objects, "parameter 'objects' is required");
        Assert.state(keys.size() == objects.size(), "size of 'keys' and 'objects' must be the same");
        Assert.state(timeout > 0, "parameter 'timeout' must be > 0");
        Assert.notNull(unit, "parameter 'unit' is required");

        List<String> jsonList = objects.stream()
                .map(JsonUtils::toJsonString)
                .toList();

        this.redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            final RedisSerializer<String> stringSerializer = this.redisTemplate.getStringSerializer();

            Iterator<String> keysIterator = keys.iterator();
            Iterator<String> jsonIterator = jsonList.iterator();

            while (keysIterator.hasNext() && jsonIterator.hasNext()) {
                final String key = keysIterator.next();
                final String json = jsonIterator.next();

                final byte[] rawKey = stringSerializer.serialize(key);
                final byte[] rawValue = stringSerializer.serialize(json);

                connection.set(rawKey, rawValue);
                connection.expire(rawKey, TimeoutUtils.toSeconds(timeout, unit));
            }

            return null;
        });
    }
}
